package cn.doitedu.demo.join;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class 常规join {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 建表，映射kafka中的ss-1
        tenv.executeSql("CREATE TABLE t1                  (\n" +
                "  uid BIGINT                                 \n" +
                "  ,event_id STRING                           \n" +
                "  ,action_time BIGINT                        \n" +
                "  ,rt1 as to_timestamp_ltz(action_time,3)    \n" +
                "  ,watermark for rt1 as rt1                  \n" +
                ") WITH (                                    \n" +
                "  'connector' = 'kafka',                    \n" +
                "  'topic' = 'a-1',                         \n" +
                "  'properties.bootstrap.servers' = 'doitedu:9092',\n" +
                "  'properties.group.id' = 'doit44_g1',      \n" +
                "  'scan.startup.mode' = 'latest-offset',    \n" +
                "  'value.format' = 'json',                  \n" +
                "  'value.fields-include' = 'EXCEPT_KEY'     \n" +
                ")");


        // 建表，映射kafka中的ss-2
        tenv.executeSql("CREATE TABLE t2                  (\n" +
                "  uid BIGINT                                \n" +
                "  ,face_status int                           \n" +
                "  ,ts bigint                                 \n" +
                "  ,rt2 as to_timestamp_ltz(ts,3)             \n" +
                "  ,watermark for rt2 as rt2                  \n" +
                ") WITH (                                    \n" +
                "  'connector' = 'kafka',                    \n" +
                "  'topic' = 'a-2',                         \n" +
                "  'properties.bootstrap.servers' = 'doitedu:9092',\n" +
                "  'properties.group.id' = 'doit44_g2',      \n" +
                "  'scan.startup.mode' = 'latest-offset',    \n" +
                "  'value.format' = 'json',                  \n" +
                "  'value.fields-include' = 'EXCEPT_KEY'     \n" +
                ")");

        tenv.executeSql(
                "select   \n" +
                "    t1.uid, \n" +
                "    t1.event_id,\n" +
                "    t1.action_time,\n" +
                "    t1.rt1,\n" +
                "    t2.face_status,\n" +
                "    t2.ts,\n" +
                "    t2.rt2\n" +
                "from t1 \n" +
                //"join t2 \n" +
                //"left join t2 \n" +
                //"right join t2 \n" +
                "full join t2 \n" +
                "on t1.uid = t2.uid").print();



    }


}
